package com.base.core.task.parser;

import com.base.core.task.annotation.ElasticJobConfig;
import com.base.core.task.autoconfigure.JobZookeeperProperties;
import com.base.core.task.enums.ElasticJobTypeEnum;
import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.executor.handler.JobProperties;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.util.StringUtils;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

@Slf4j
public class ElasticJobConfParser implements ApplicationListener<ApplicationReadyEvent> {

    private JobZookeeperProperties jobZookeeperProperties;
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    public ElasticJobConfParser(JobZookeeperProperties jobZookeeperProperties, ZookeeperRegistryCenter zookeeperRegistryCenter) {
        this.jobZookeeperProperties=jobZookeeperProperties;
        this.zookeeperRegistryCenter=zookeeperRegistryCenter;
    }


    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        try {
            //应用服务器都已经启动,初始化完成事件
            ApplicationContext applicationContext=event.getApplicationContext();
            Map<String,Object> beanMap=applicationContext.getBeansWithAnnotation(ElasticJobConfig.class);
            for(Iterator<?> it = beanMap.values().iterator(); it.hasNext();){
                Object confBean=it.next();
                Class<?> clazz=confBean.getClass();
                if(clazz.getName().indexOf("$")>0){
                    String className=clazz.getName();
                    clazz=Class.forName(className.substring(0,className.indexOf("$")));
                }
                //获取接口类型用于判断是什么类型的任务
                String jobTypeName=clazz.getInterfaces()[0].getSimpleName();
                //获取配置项
                ElasticJobConfig elasticJobConfig= clazz.getAnnotation(ElasticJobConfig.class);
                String jobClass=clazz.getName();
                String jobName=this.jobZookeeperProperties.getNamespace()+"."+elasticJobConfig.name();
                String cron= elasticJobConfig.cron();
                String shardingItemParameters = elasticJobConfig.shardingItemParameters();
                String description = elasticJobConfig.description();
                String jobParameter = elasticJobConfig.jobParameter();
                String jobExceptionHandler = elasticJobConfig.jobExceptionHandler();
                String executorServiceHandler = elasticJobConfig.executorServiceHandler();

                String jobShardingStrategyClass = elasticJobConfig.jobShardingStrategyClass();
                String eventTraceRdbDataSource = elasticJobConfig.eventTraceRdbDataSource();
                String scriptCommandLine = elasticJobConfig.scriptCommandLine();

                boolean failover = elasticJobConfig.failover();
                boolean misfire = elasticJobConfig.misfire();
                boolean overwrite = elasticJobConfig.overwrite();
                boolean disabled = elasticJobConfig.disabled();
                boolean monitorExecution = elasticJobConfig.monitorExecution();
                boolean streamingProcess = elasticJobConfig.streamingProcess();

                int shardingTotalCount = elasticJobConfig.shardingTotalCount();
                int monitorPort = elasticJobConfig.monitorPort();
                int maxTimeDiffSeconds = elasticJobConfig.maxTimeDiffSeconds();
                int reconcileIntervalMinutes = elasticJobConfig.reconcileIntervalMinutes();

                // 创建JobCoreConfiguration,先把当当网的esjob的相关configuration
                //先把当当网的esjob的相关configuration
                JobCoreConfiguration coreConfiguration = JobCoreConfiguration
                        .newBuilder(jobName, cron, shardingTotalCount)
                        .shardingItemParameters(shardingItemParameters)
                        .description(description)
                        .failover(failover)
                        .jobParameter(jobParameter)
                        .misfire(misfire)
                        .jobProperties(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), jobExceptionHandler)
                        .jobProperties(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(), executorServiceHandler)
                        .build();

                // 根据实现类类型创建不同类型的JOB
                //我到底要创建什么样的任务
                JobTypeConfiguration typeConfiguration = null;
                if (ElasticJobTypeEnum.SIMPLE.getType().equals(jobTypeName)) {
                    typeConfiguration = new SimpleJobConfiguration(coreConfiguration, jobClass);
                } else if (ElasticJobTypeEnum.DATAFLOW.getType().equals(jobTypeName)) {
                    typeConfiguration = new DataflowJobConfiguration(coreConfiguration, jobClass, streamingProcess);
                } else if (ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                    typeConfiguration = new ScriptJobConfiguration(coreConfiguration, scriptCommandLine);
                }
                // 创建LiteJobConfiguration
                LiteJobConfiguration jobConfiguration = LiteJobConfiguration
                        .newBuilder(typeConfiguration)
                        .overwrite(overwrite)
                        .disabled(disabled)
                        .monitorPort(monitorPort)
                        .monitorExecution(monitorExecution)
                        .maxTimeDiffSeconds(maxTimeDiffSeconds)
                        .jobShardingStrategyClass(jobShardingStrategyClass)
                        .reconcileIntervalMinutes(reconcileIntervalMinutes)
                        .build();

                // 创建一个Spring的beanDefinition
                BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
                factory.setInitMethodName("init");
                //多例
                factory.setScope("prototype");
                //1、 添加bean构造参数,相当于添加自己的真实的任务实现类
                if (!ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                    factory.addConstructorArgValue(confBean);
                }
                //2、 添加注册中心
                factory.addConstructorArgValue(zookeeperRegistryCenter);
                //3、 添加LiteJobConfiguration
                factory.addConstructorArgValue(jobConfiguration);

                //4、 如果有eventTraceRdbDataSource，也需进行添加
                if (StringUtils.hasText(eventTraceRdbDataSource)) {
                    BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(JobEventRdbConfiguration.class);
                    rdbFactory.addConstructorArgReference(eventTraceRdbDataSource);
                    factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
                }

                // 5、添加监听
                List<?> elasticJobListeners = getTargetElasticJobListeners(elasticJobConfig);
                factory.addConstructorArgValue(elasticJobListeners);

                 //接下来就是把factory也就是springJobScheduler注入到spring容器中
                // 将SpringJobScheduler注入到Spring容器中
                DefaultListableBeanFactory listableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();
                String registerBeanName = jobName + "SpringJobScheduler";
                listableBeanFactory.registerBeanDefinition(registerBeanName, factory.getBeanDefinition());
                SpringJobScheduler springJobScheduler = (SpringJobScheduler) applicationContext.getBean(registerBeanName);
                springJobScheduler.init();
                log.info("启动elastic-job作业：", jobName);
            }
            log.info("共计启动启动elastic-job作业:{}个 ", beanMap.values().size());
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            log.error("elasticjob 启动异常，系统强制退出",e);
            System.exit(1);
        }
    }


    private List<BeanDefinition> getTargetElasticJobListeners(ElasticJobConfig conf) {
        List<BeanDefinition> result = new ManagedList<>(2);
        String listeners = conf.listener();
        if (StringUtils.hasText(listeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(listeners);
            factory.setScope("prototype");
            result.add(factory.getBeanDefinition());
        }

        String distributedListeners = conf.distributedListener();
        long startedTimeoutMilliseconds = conf.startedTimeoutMilliseconds();
        long completedTimeoutMilliseconds = conf.completedTimeoutMilliseconds();

        if (StringUtils.hasText(distributedListeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(distributedListeners);
            factory.setScope("prototype");
            factory.addConstructorArgValue(startedTimeoutMilliseconds);
            factory.addConstructorArgValue(completedTimeoutMilliseconds);
            result.add(factory.getBeanDefinition());
        }
        return result;
    }

}
